package com.mjf.transformation

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

/**
 * connect：功能与union类似，将两个流（union支持两个或以上）合并为一个流，但区别在于connect不要求数据类型一致
 *
 * connect 之后需要进行其他的操作将 ConnectedStreams 转换成 DataStream
 */
object CoMapDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    val source1: DataStream[(String, Int)] = env.fromCollection(List(("tina", 23),("paul", 30)))

    val source2: DataStream[(String, String)] = env.fromCollection(List(("lucky", "female"),("paul", "male")))

    source1.print("source1")
    source2.print("source2")

    val connectStream: ConnectedStreams[(String, Int), (String, String)] = source1.connect(source2)

    val result: DataStream[String] = connectStream.map(new MyCoMapFunction)

    result.print("result")

    env.execute(CoMapDemo.getClass.getName)

  }
}

class MyCoMapFunction extends CoMapFunction[(String, Int), (String, String), String] {
  // 处理第一条流的数据
  override def map1(value: (String, Int)): String = value._1 + "的年龄：" + value._2

  // 处理第二条流的数据
  override def map2(value: (String, String)): String = value._1 + "的性别：" + value._2
}
